package com.cloudwise.isop.drools;

import com.cloudwise.isop.drools.dao.AlertRuleDAO;
import com.cloudwise.isop.drools.listener.ComsumerListener;
import com.cloudwise.isop.drools.pojo.AlertRulePO;
import com.cloudwise.isop.drools.utils.AlertDataUtils;
import com.cloudwise.isop.drools.utils.CommonUtils;
import com.cloudwise.isop.drools.utils.ConfigUtil;
import com.cloudwise.isop.drools.utils.ConfigUtils;
import com.typesafe.config.Config;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.RegexSubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
import org.drools.core.impl.InternalKnowledgeBase;
import org.drools.core.impl.KnowledgeBaseFactory;
import org.kie.api.definition.KiePackage;
import org.kie.api.io.Resource;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.internal.builder.KnowledgeBuilder;
import org.kie.internal.builder.KnowledgeBuilderFactory;
import org.kie.internal.io.ResourceFactory;

import java.io.Reader;
import java.io.StringReader;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

/**
 * drools demo 入口文件， 使用字符串规则
 * @Author jeffse.jiang
 */
@Slf4j
public class StringRuleEntrance {
    private static KnowledgeBuilder kbuilder = KnowledgeBuilderFactory.newKnowledgeBuilder();
    private static Collection<KiePackage> pkgs;
    private static InternalKnowledgeBase kbase = KnowledgeBaseFactory.newKnowledgeBase();
    private static KieSession ksession;

    public static void main(String[] args) throws PulsarClientException {
        // 初始化规则
        initRules();
        // Check the builder for errors
        if ( kbuilder.hasErrors() ) {
            System.out.println( kbuilder.getErrors().toString() );
            throw new RuntimeException( "Unable to compile drl\"." );
        }

        // get the compiled packages (which are serializable)
        pkgs = kbuilder.getKnowledgePackages();

        // add the packages to a knowledgebase (deploy the knowledge packages).
        kbase.addPackages( pkgs );

        ksession = kbase.newKieSession();
        Config config = ConfigUtils.getConfig();

        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(config.getString("pulsar.serviceUrl")).build();
        Pattern someTopicsInNamespace = Pattern.compile(config.getString("pulsar.topic"));
        pulsarClient.newConsumer().messageListener(new ComsumerListener(ksession))
                .subscriptionName(config.getString("pulsar.groupName"))
                .topicsPattern(someTopicsInNamespace)
                .ackTimeout(10, TimeUnit.SECONDS)
                .subscriptionType(SubscriptionType.Shared)
                .subscribe();
    }

    public static List<AlertRulePO> initRules(){
        AlertRuleDAO alertRuleDAO = CommonUtils.getInstance(AlertRuleDAO.class);
        List<AlertRulePO> alertRulePOS = alertRuleDAO.getList();
        AlertDataUtils.init(alertRulePOS);
        alertRulePOS.forEach(item->{
            Resource myResource = ResourceFactory.newReaderResource((Reader) new StringReader(item.getRule()));
            kbuilder.add(myResource, ResourceType.DRL);
        });
        return alertRulePOS;
    }
}
